package com.atguigu.gmall.realtime.app

import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util
import java.util.Date

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{DauInfo, OrderDetail, OrderInfo, OrderWide}
import com.atguigu.gmall.realtime.utils.{MyEsUtils, MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * 日活宽表
 * 1.获取环境对象
 * 2.读取redis中的offset  * 2
 * 3.从kafka中消费数据   * 2
 * 4.获取当前消费的offset   * 2
 * 5.处理数据
 *     5.1转换数据结构
 *     5.2 维度关联
 *     5.3 双流join
 *6.写入es
 *7.提交offset
 */
object DwdOrderApp {
  def main(args: Array[String]): Unit = {
    //1.获取环境对象
    val conf: SparkConf = new SparkConf().setAppName("dwd_order_app").setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(5))

    //2.读取redis中的offset
    val orderInfoTopicName = "DWD_ORDER_INFO_I_1118"
    val orderDetailTopicName = "DWD_ORDER_DETAIL_I_1118"
    val group = "DWD_ORDER_GROUP"

    val orderInfoOffsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(orderInfoTopicName,group)
    val orderDetailOffsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(orderDetailTopicName,group)


    // 3.从kafka中消费数据
    var orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderInfoOffsets != null && orderInfoOffsets.nonEmpty){
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,orderInfoTopicName,group,orderInfoOffsets)
    }else{
      orderInfoKafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,orderInfoTopicName,group)
    }
    var orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(orderDetailOffsets != null && orderDetailOffsets.nonEmpty){
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,orderDetailTopicName,group,orderDetailOffsets)
    }else{
      orderDetailKafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,orderDetailTopicName,group)
    }


    // 4.获取当前消费的offset
    var orderInfoOffetRanges: Array[OffsetRange] = null
    val orderInfoOffsetDStream: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform(
      rdd => {
        orderInfoOffetRanges  = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    var orderDetailOffetRanges: Array[OffsetRange] = null
    val orderDetailOffsetDStream: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform(
      rdd => {
        orderDetailOffetRanges  = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    //5.数据处理
    //5.1转换结构
    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderInfo: OrderInfo = JSON.parseObject(value, classOf[OrderInfo])
        orderInfo
      }
    )
    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val orderDetail: OrderDetail = JSON.parseObject(value, classOf[OrderDetail])
        orderDetail
      }
    )
    //5.2维度关联
    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      orderInfoIter => {
        val jedis: Jedis = MyRedisUtils.getJedis()
        val orderInfoes: List[OrderInfo] = orderInfoIter.toList.map(
          orderInfo => {
            //关联用户维度
            val userId: Long = orderInfo.user_id
            val userRedisKey: String = s"DIM:USER_INFO:$userId"
            val userJson: String = jedis.get(userRedisKey)
            val userJSONObj: JSONObject = JSON.parseObject(userJson)
            //提取维度数据
            val gender: String = userJSONObj.getString("gender")
            val birthday: String = userJSONObj.getString("birthday")

            //换算年龄
            val birthdayLd: LocalDate = LocalDate.parse(birthday)
            val nowLd: LocalDate = LocalDate.now()
            val period: Period = Period.between(birthdayLd, nowLd)
            val age: Int = period.getYears
            //补充维度
            orderInfo.user_gender = gender
            orderInfo.user_age = age

            //关联地区维度
            val provinceId: Long = orderInfo.province_id
            val provinceRedisKey: String = s"DIM:BASE_PROVINCE:$provinceId"
            val provinceStr: String = jedis.get(provinceRedisKey)
            val provinceJsonObj: JSONObject = JSON.parseObject(provinceStr)

            //提取维度数据
            val provinceName: String = provinceJsonObj.getString("name")
            val provinceAreaCode: String = provinceJsonObj.getString("area_code")
            val provinceIsoCode: String = provinceJsonObj.getString("iso_code")
            val provinceIso3166: String = provinceJsonObj.getString("iso_3166_2")

            //补充维度
            orderInfo.province_name = provinceName
            orderInfo.province_area_code = provinceAreaCode
            orderInfo.province_iso_code = provinceIsoCode
            orderInfo.province_3166_2_code = provinceIso3166

            //处理日期字段
            val create_time: String = orderInfo.create_time
            val createDtHr: Array[String] = create_time.split(" ")
            orderInfo.create_date = createDtHr(0)
            orderInfo.create_hour = createDtHr(1).split(":")(0)

            orderInfo
          }
        )
        orderInfoes
        MyRedisUtils.close(jedis)
        orderInfoes.toIterator
      }
    )
    //orderInfoDimDStream.print(100)


    //5.3 双流Join
    // SQL:  join操作很好实现
    // 流Join :
    // 内连接:
    //      join  ：只取交集
    // 外连接:
    //      leftOuterJoin : 左表所有 + 右表匹配
    //      rightOuterJoin: 左表匹配 + 右表所有
    //                      其实在左外或者右外中，更重要的是确认哪张表是主表，哪张表是从表.
    //      fullOuterJoin:  全外连 , 两表的数据都要.

    // 理论上order_info 和 order_detail表的数据都能join成功.
    // 因此在SQL层面使用哪种join结果都是一样的。
    // 但是在流层面, 我们不敢保证两个表中对应的数据同时出现到同一个批次中. 而流join的特点是只有同一批次的数据才能Join成功.
    // 如果数据出现延迟， 不能保证两个表对应的数据进入到同一个批次，就会导致数据不能完全join成功
    // 我们可以接受因为数据延迟导致最终统计结果也出现延迟.
    // 但是我们不能接受因为数据延迟导致数据丢失.
    // 如何解决??

    // 流join要求的数据格式是kv的
    val orderInfoKVDStream: DStream[(Long, OrderInfo)] = orderInfoDimDStream.map( orderInfo => (orderInfo.id ,orderInfo))
    val orderDetailKVDStream: DStream[(Long, OrderDetail)] = orderDetailDStream.map(orderDetail => (orderDetail.order_id , orderDetail))

    //val joinDStream: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoKVDStream.join(orderDetailKVDStream)
    //joinDStream.print(10000)

    //fullOuterJoin
    val orderJoinDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] =
      orderInfoKVDStream.fullOuterJoin(orderDetailKVDStream)

    //orderJoinDStream.print(10000)
    val orderWideDStream: DStream[OrderWide] = orderJoinDStream.mapPartitions(
      orderJoinIter => {
        val orderWides: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        val jedis: Jedis = MyRedisUtils.getJedis()
        for ((key, (orderInfoOp, orderDetailOp)) <- orderJoinIter) {
          // orderInfo有
          if (orderInfoOp.isDefined) {
            val orderInfo: OrderInfo = orderInfoOp.get
            //orderDetail有
            if (orderDetailOp.isDefined) {
              val orderDetail: OrderDetail = orderDetailOp.get
              //封装成orderWide
              val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)
              //存储到结果集合中
              orderWides.append(orderWide)
            }
            //orderInfo写缓存
            //type:  string
            //key :   ORDERJOIN:ORDERINFO:ID
            //value : json
            //写入API: set
            //读取API: get
            //是否过期: 24 * 3600
            val orderInfoRedisKey: String = s"ORDERJOIN:ORDERINFO:${orderInfo.id}"
            jedis.setex(orderInfoRedisKey, 3600 * 24, JSON.toJSONString(orderInfo, new SerializeConfig(true)))

            //orderInfo读缓存
            // type :  set
            // key :   ORDERJOIN:ORDERDETAIL:ORDERID
            // value : json
            // 写入API: sadd
            // 读取API: smembers
            // 是否过期: 24 * 3600

            val orderDetailRedisKey: String = s"ORDERJOIN:ORDERDETAIL:${orderInfo.id}"
            val orderDetails: util.Set[String] = jedis.smembers(orderDetailRedisKey)
            if (orderDetails != null && orderDetails.size() > 0) {
              import scala.collection.JavaConverters._
              for (orderDetailJson <- orderDetails.asScala) {
                val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
                val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)
                orderWides.append(orderWide)
              }
            }
          } else {
            // orderInfo没有,orderDetail有
            val orderDetail: OrderDetail = orderDetailOp.get
            //读缓存
            val orderInfoRedisKey: String = s"ORDERJOIN:ORDERINFO:${orderDetail.order_id}"
            val orderInfoJson: String = jedis.get(orderInfoRedisKey)
            if (orderInfoJson != null && orderInfoJson.size > 0) {
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
              val orderWide: OrderWide = new OrderWide(orderInfo, orderDetail)
              orderWides.append(orderWide)
            } else {
              //写缓存
              val orderDetailRedisKey: String = s"ORDERJOIN:ORDERDETAIL:${orderDetail.order_id}"
              jedis.sadd(orderDetailRedisKey, JSON.toJSONString(orderDetail, new SerializeConfig(true)))
              jedis.expire(orderDetailRedisKey, 24 * 3600)
            }
          }

        }
        MyRedisUtils.close(jedis)
        orderWides.toIterator
      }
    )
    //orderWideDStream.print(100000)
    orderWideDStream.foreachRDD(
      rdd =>{
        rdd.foreachPartition(
          orderWideIter =>{
            val sources: List[(String, OrderWide)] = orderWideIter.map(orderWide => (orderWide.detail_id,orderWide)).toList
            if(sources.size > 0){
              val orderWide: OrderWide = sources.head._2
              val dt: String = orderWide.create_date
              val indexName : String = s"gmall_order_wide_$dt"
              MyEsUtils.save(indexName,sources)
            }
          }
        )
        //7.提交offset
        MyOffsetUtils.saveOffset(orderInfoTopicName,group,orderInfoOffetRanges)
        MyOffsetUtils.saveOffset(orderDetailTopicName,group,orderDetailOffetRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()


  }
}
